package io.github.junxworks.junx.stream.handler;

import io.github.junxworks.junx.stream.constants.DefaultParams;
import io.github.junxworks.junx.stream.constants.DefaultTopics;
import io.github.junxworks.junx.event.EventChannel;
import io.github.junxworks.junx.event.EventContext;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public abstract class AbstractBatchDataDataHandler extends AbstractDataHandler {

    protected int batchSize = 1000;

    protected List<EventContext> events = new ArrayList<EventContext>(batchSize);

    public int getBatchSize() {
        return batchSize;
    }

    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    @Override
    public void handleEvent(EventContext event, EventChannel channel) throws Exception {
        events.add(event);
        if (event.isEndOfBatch() || events.size() >= batchSize) {
            incrCount();
            try {
                List<Object> dataList = events.stream().map(e -> e.getData(DefaultParams.PARAM_DATA)).collect(ArrayList::new, ArrayList::add, ArrayList::addAll);
                List<Object> resList = handleEvents(dataList, globalParams);
                if (shouldPassEvent() && resList != null && !resList.isEmpty()) {
                    for (Object o : resList) {
                        EventContext ctx = new EventContext(outputTopic);
                        ctx.setData(DefaultParams.PARAM_DATA, o);
                        channel.publish(outputTopic, ctx);
                    }
                }
            } catch (Exception e) {
                event.setData(DefaultParams.PARAM_EXCEPTION, e);
                event.setData(DefaultParams.PARAM_GLOBAL_PARAMS, globalParams);
                channel.publish(DefaultTopics.TOPIC_EXCEPTION, event);
            } finally {
                decrCount();
                events.clear();
            }
        }
    }

    public abstract List<Object> handleEvents(List<Object> dataList, Map<String, Object> globalParams) throws Exception;

    @Override
    public boolean isDone() {
        return super.isDone() && events.isEmpty();
    }
}
